package com.weishao.example;

import java.util.List;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.LinkedList;
import org.apache.commons.dbcp2.BasicDataSource;
import org.springframework.util.StopWatch;
import com.weishao.dbchange.IDatabaseChangeCaculator;
import com.weishao.dbchange.IDatabaseRowHandler;
import com.weishao.dbchange.RecordChangeType;
import com.weishao.dbchange.ChangeCaculatorService;
import com.weishao.dbchange.pojo.TaskParamBean;
import com.weishao.dbwriter.DatabaseWriterFactory;
import com.weishao.dbwriter.IDatabaseWriter;

public class DbchangeExampleApplication {

	private static int BATCH_SIZE = 10000;

	public static void main(String[] args) {
		BasicDataSource dataSource = new BasicDataSource();
		dataSource.setDriverClassName("org.postgresql.Driver");
		dataSource.setUrl("jdbc:postgresql://172.17.207.151:5432/study");
		dataSource.setUsername("study");
		dataSource.setPassword("123456");
		dataSource.setInitialSize(5);
		dataSource.setMaxIdle(5);
		dataSource.setMinIdle(2);

		String outSchemaName = "private";
		String outTableName = "ofuser_diff";
		String flagFieldName = "diff_type";

		TaskParamBean.TaskParamBeanBuilder taskBuilder = TaskParamBean.builder();
		taskBuilder.oldDataSource(dataSource);
		taskBuilder.oldSchemaName("private");
		taskBuilder.oldTableName("ofuser_old");
		taskBuilder.newDataSource(dataSource);
		taskBuilder.newSchemaName("private");
		taskBuilder.newTableName("ofuser_new");

		IDatabaseWriter writer = DatabaseWriterFactory.createDatabaseWriter(dataSource);
		writer.prepareWrite(outSchemaName, outTableName);

		StopWatch watch = new StopWatch();
		watch.start("watcher");
		
		IDatabaseChangeCaculator changeCaculator = new ChangeCaculatorService();
		changeCaculator.setFetchSize(5000);
		changeCaculator.setRecordIdentical(false);
		
		changeCaculator.executeCalculate(taskBuilder.build(), new IDatabaseRowHandler() {
			
			private long count=0;
			private List<Object[]> cache = new LinkedList<Object[]>();

			@Override
			public void handle(List<String> fields, Object[] record, RecordChangeType flag) {
				// 这里处理输出的每一条变化量记录数据:处理的方法为：
				// (1)向对记录结果数组record扩展一个元素，来存储变化状态
				// (2)然后将该条记录数据缓存到内存cache对象中，
				// (3)当内存cache缓冲满一个批次大小后，将其批量写入结果表中。
				Object[] item = Arrays.copyOf(record, record.length + 1);
				if (flag == RecordChangeType.VALUE_INSERT) {
					item[item.length - 1] = "new";
				} else if (flag == RecordChangeType.VALUE_CHANGED) {
					item[item.length - 1] = "changed";
				} else if (flag == RecordChangeType.VALUE_DELETED) {
					item[item.length - 1] = "deleted";
				} else {
					// 因前面执行了changeCaculator.setRecordIdentical(false);
					// 其实代码不会执行到这里的。
					return;
				}
				
				cache.add(item);
				if (cache.size() >= BATCH_SIZE) {
					doSave(fields);
				}
			}

			@Override
			public void destroy(List<String> fields) {
				if (cache.size() > 0) {
					doSave(fields);
				}
				
				System.out.println("total handle count: " + this.count);
			}

			private void doSave(List<String> fields) {
				List<String> fieldNames = new ArrayList<String>(fields);
				fieldNames.add(flagFieldName);
				long ret = writer.write(fieldNames, cache);
				System.out.println("handle result count: " + ret);
				count += ret;
				cache.clear();
			}

		});

		watch.stop();
		System.out.println("Total elipse :" + watch.getTotalTimeSeconds() + " s");
		System.out.println(watch.prettyPrint());
	}
}
